package cn._51doit.flink.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 先keyBy，然后再按照EventTime(数据里所携带的时间)划分滚动窗口
 *
 * 按照EventTime划分窗口，是按照数据中所携带的是生成窗口
 * 窗口的长度就是你指定的时间，窗口的起始时间，结束时间是对齐的，即可以整除窗口的长度
 *
 * 设置延迟时间（2秒），让窗口可以延迟触发（数据运行乱序进入窗口，并且可以延迟一定时间在触发）
 *
 */
public class EventTimeTumblingWindowDemoOldApi3 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1000,spark,3
        //3000,hadoop,5
        //4000,spark,1
        //9000,spark,1
        //9998,spark,1
        //9999,spark,1
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //提取数据中的eventTime
        //BoundedOutOfOrdernessTimestampExtractor（允许数据中的eventTime是乱序的，有时间边界范围的时间提取器）
        //assignTimestampsAndWatermarks该方法仅是提取数据中的EventTime，提取完返回的数据没有任何改变
        SingleOutputStreamOperator<String> dataStreamWithWaterMark = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(String element) {
                String[] fields = element.split(",");
                return Long.parseLong(fields[0]);
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataStreamWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                String word = fields[1];
                int count = Integer.parseInt(fields[2]);
                return Tuple2.of(word, count);
            }
        });

        //先KeyBy
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(tp -> tp.f0);

        //按照EventTime滚动的窗口，窗口长度为10秒
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = window.sum(1);

        res.print();

        env.execute();

    }
}
